package com.shen.spark.kafka;

import com.alibaba.fastjson.JSONObject;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Random;


public class KafkaProducer extends Thread{

    private String topic;
    private String user;

    private Producer<Integer, String> producer;

    public KafkaProducer(String topic, String user) {
        this.topic = topic;
        this.user = user;

        Properties properties = new Properties();

        properties.put("metadata.broker.list",KafkaProperties.BROKER_LIST);
        properties.put("serializer.class","kafka.serializer.StringEncoder");
        properties.put("request.required.acks","1");

        producer = new Producer<Integer, String>(new ProducerConfig(properties));
    }


    @Override
    public void run() {

        int messageNo = 1;
        Random random = new Random();

        while(true) {
            int orderId = random.nextInt(20);
            Map map = new HashMap();
            map.put("time", "123");
            map.put("userid", user+messageNo);
            map.put("courseid","courseid");
            map.put("fee", "fee");
            map.put("flag", "flag");
            map.put("orderid",orderId);

            String json = JSONObject.toJSONString(map);
            producer.send(new KeyedMessage<Integer, String>(topic, json));
            System.out.println("Sent: " + json);

            messageNo ++ ;

            try{
                Thread.sleep(2000);
            } catch (Exception e){
                e.printStackTrace();
            }
        }

    }
}